package com.yanggu.flink.datastream_api.checkpoint

import org.apache.flink.api.common.restartstrategy.RestartStrategies
import org.apache.flink.api.common.restartstrategy.RestartStrategies.RestartStrategyConfiguration
import org.apache.flink.runtime.state.storage.FileSystemCheckpointStorage
import org.apache.flink.streaming.api.CheckpointingMode
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment

/**
 * https://www.icode9.com/content-4-770147.html
 */
object CheckpointDemo {

  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment

    //开启checkpoint配置, 两次checkpoint首首之间的间隔
    env.enableCheckpointing(5000L)
    //配置重启策略
    env.setRestartStrategy(RestartStrategies.fixedDelayRestart(20, 5000L))
    val checkpointConfig = env.getCheckpointConfig
    //checkpoint的url路径
    checkpointConfig.setCheckpointStorage("hdfs://hadoop100:8020/gmall-flink/checkpoint")
    //checkpoint的模式: 精确一致性
    checkpointConfig.setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE)
    //checkpoint超时时间
    checkpointConfig.setCheckpointTimeout(10000L)
    //checkpoint最大次数
    checkpointConfig.setMaxConcurrentCheckpoints(2)
    //第一次checkpoint的尾和第二次checkpoint的首之间的时间间隔
    checkpointConfig.setMinPauseBetweenCheckpoints(3000)
  }

}
